package beautiful.butterfly.drds.data_exchange.report;


import beautiful.butterfly.drds.data_exchange.constant.State;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.Validate;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;


/**
 * 所有的状态及统计信息交互类，job、taskGroup、task等的消息汇报都走该类
 */
public class Message implements Cloneable
{
    /**
     * task给job的信息 *
     */
    Map<String, List<String>> taskIdToMessageListMap;
    /**
     * 所有的数值key-value对 *
     */
    private Map<String, Number> keyToNumberMap;
    /**
     * 运行状态 *
     */
    private State state;
    /**
     * 异常记录 *
     */
    private Throwable throwable;
    /**
     * 记录的timestamp *
     */
    private long timestamp;

    public Message()
    {
        this.init();
    }

    public synchronized void reset()
    {
        this.init();
    }

    private void init()
    {
        this.keyToNumberMap = new ConcurrentHashMap<String, Number>();
        this.state = State.running;
        this.throwable = null;
        this.taskIdToMessageListMap = new ConcurrentHashMap<String, List<String>>();
        this.timestamp = System.currentTimeMillis();
    }

    public Map<String, Number> getKeyToNumberMap()
    {
        return this.keyToNumberMap;
    }

    public State getState()
    {
        return this.state;
    }

    public synchronized void setState(State state)
    {
        setState(state, false);
    }

    public synchronized void setState(State state, boolean isForce)
    {
        if (!isForce && this.state.equals(State.failed))
        {
            return;
        }

        this.state = state;
    }

    public Throwable getThrowable()
    {
        return this.throwable;
    }

    public void setThrowable(Throwable throwable)
    {
        setThrowable(throwable, false);
    }

    public synchronized String getThrowableMessage()
    {
        return this.throwable == null ? "" : this.throwable.getMessage();
    }

    public synchronized void setThrowable(Throwable throwable, boolean isForce)
    {
        if (isForce)
        {
            this.throwable = throwable;
        } else
        {
            this.throwable = this.throwable == null ? throwable : this.throwable;
        }
    }

    public long getTimestamp()
    {
        return this.timestamp;
    }

    public void setTimestamp(long timestamp)
    {
        this.timestamp = timestamp;
    }

    public Map<String, List<String>> getTaskIdToMessageListMap()
    {
        return this.taskIdToMessageListMap;
    }

    public List<String> getMessage(final String key)
    {
        return taskIdToMessageListMap.get(key);
    }

    public synchronized void addMessage(final String key, final String value)
    {
        Validate.isTrue(StringUtils.isNotBlank(key), "增加message的key不能为空");
        List list = this.taskIdToMessageListMap.get(key);
        if (null == list)
        {
            list = new ArrayList<String>();
            this.taskIdToMessageListMap.put(key, list);
        }

        list.add(value);
    }

    public synchronized Long getLong(final String key)
    {
        Number value = this.keyToNumberMap.get(key);

        return value == null ? 0 : value.longValue();
    }

    public synchronized void setLong(final String key, final long value)
    {
        Validate.isTrue(StringUtils.isNotBlank(key), "设置counter的key不能为空");
        this.keyToNumberMap.put(key, value);
    }

    public synchronized Double getDouble(final String key)
    {
        Number value = this.keyToNumberMap.get(key);

        return value == null ? 0.0d : value.doubleValue();
    }

    public synchronized void setDouble(final String key, final double value)
    {
        Validate.isTrue(StringUtils.isNotBlank(key), "设置counter的key不能为空");
        this.keyToNumberMap.put(key, value);
    }

    public synchronized void increase(final String key, final long deltaValue)
    {
        Validate.isTrue(StringUtils.isNotBlank(key), "增加counter的key不能为空");

        long value = this.getLong(key);

        this.keyToNumberMap.put(key, value + deltaValue);
    }

    @Override
    public Message clone()
    {
        Message message = new Message();

        /**
         * clone keyToNumberMap
         */
        if (this.keyToNumberMap != null)
        {
            for (Entry<String, Number> entry : this.keyToNumberMap.entrySet())
            {
                String key = entry.getKey();
                Number value = entry.getValue();
                if (value instanceof Long)
                {
                    message.setLong(key, (Long) value);
                } else if (value instanceof Double)
                {
                    message.setDouble(key, (Double) value);
                }
            }
        }

        message.setState(this.state, true);
        message.setThrowable(this.throwable, true);
        message.setTimestamp(this.timestamp);

        /**
         * clone taskIdToMessageListMap
         */
        if (this.taskIdToMessageListMap != null)
        {
            for (final Entry<String, List<String>> entry : this.taskIdToMessageListMap.entrySet())
            {
                String key = entry.getKey();
                List value = new ArrayList()
                {{
                    addAll(entry.getValue());
                }};
                message.getTaskIdToMessageListMap().put(key, value);
            }
        }

        return message;
    }

    public synchronized Message merge(final Message message)
    {
        if (message == null)
        {
            return this;
        }

        /**
         * counter的合并，将otherComm的值累加到this中，不存在的则创建
         * 同为long
         */
        for (Entry<String, Number> entry : message.getKeyToNumberMap().entrySet())
        {
            String key = entry.getKey();
            Number value = entry.getValue();
            if (value == null)
            {
                continue;
            }

            Number number = this.keyToNumberMap.get(key);
            if (number == null)
            {
                number = value;
            } else
            {
                if (number instanceof Long && value instanceof Long)
                {
                    number = number.longValue() + value.longValue();
                } else
                {
                    number = number.doubleValue() + number.doubleValue();
                }
            }

            this.keyToNumberMap.put(key, number);
        }

        // 合并state
        mergeState(message);

        /**
         * 合并throwable，当this的throwable为空时，
         * 才将otherComm的throwable合并进来
         */
        this.throwable = this.throwable == null ? message.getThrowable() : this.throwable;

        /**
         * timestamp是整个一次合并的时间戳，单独两两communication不作合并
         */

        /**
         * message的合并采取求并的方式，即全部累计在一起
         */
        for (Entry<String, List<String>> entry : message.getTaskIdToMessageListMap().entrySet())
        {
            String key = entry.getKey();
            List<String> list = this.taskIdToMessageListMap.get(key);
            if (list == null)
            {
                list = new ArrayList<String>();
                this.taskIdToMessageListMap.put(key, list);
            }

            list.addAll(entry.getValue());
        }

        return this;
    }

    /**
     * 合并state，优先级： (Failed | Killed) > Running > Success
     * 这里不会出现 Killing 状态，killing 状态只在 job 自身状态上才有.
     */
    public synchronized State mergeState(final Message message)
    {
        State state = this.getState();
        if (message == null)
        {
            return state;
        }

        if (this.state == State.failed || message.getState() == State.failed
                || this.state == State.killed || message.getState() == State.killed)
        {
            state = State.failed;
        } else if (this.state.isRunning() || message.state.isRunning())
        {
            state = State.running;
        }

        this.setState(state);
        return state;
    }

    public synchronized boolean isFinished()
    {
        return this.state == State.succeeded || this.state == State.failed
                || this.state == State.killed;
    }

}
